/**
 * This class processes the incoming requests to the coordinator in FIFO order.
 */
package com.comp6231.coordinator;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.NoSuchElementException;

import com.comp6231.common.Request;

/**
 * @author Farzana Alam
 * 
 */
public class FIFOSubsystem implements Runnable {

	/*
	 * Key: Replica Name; Value: [0] => Host Name; [1] => Data UDP Port; [2] =>
	 * Control UDP Port;
	 */
	private Hashtable<String, String[]> coordinatorReplicaList;
	private Coordinator myCoordinator;

	private LinkedList<Request> incomingRequestQueue;

	public FIFOSubsystem(Hashtable<String, String[]> coordinatorReplicaList,
			Coordinator myCoordinator, LinkedList<Request> incomingRequestQueue) {
		super();
		this.coordinatorReplicaList = coordinatorReplicaList;
		this.myCoordinator = myCoordinator;
		this.incomingRequestQueue = incomingRequestQueue;
	}

	@Override
	public void run() {
		while (true) {
			Request request;
			try {
				request = incomingRequestQueue.removeFirst();
			} catch (NoSuchElementException nse) {
				continue;
			}

			// if this coordinator is not the leader, it wont have to forward
			// FIFO requests to other replicas
			if (!myCoordinator.isLeader) {
				request.setRemotelyProcessd(true);
				continue;
			}

			Enumeration<String> keys = coordinatorReplicaList.keys();
			StringBuilder[] replyMessage = new StringBuilder[2];
			int i = 0;

			// forwarding FIFO requests to the other replicas
			while (keys.hasMoreElements()) {
				final String key = keys.nextElement();
				if (key.equalsIgnoreCase(myCoordinator.myReplicaName))
					continue;
				replyMessage[i] = new StringBuilder("success");
				String[] replicaDetails = coordinatorReplicaList.get(key);
				final String hostName = replicaDetails[0];
				final String UDPPort = replicaDetails[1];

				UDPClient client = new UDPClient(request, hostName, UDPPort,
						replyMessage[i++], key);
				new Thread(client).start();
			}

			// If all others replicas are dead continue with the next request
			if (coordinatorReplicaList.keySet().size() < 2) {
				request.setRemotelyProcessd(true);
				continue;

			} else {

				// do not process the next request unless Success
				// message received from both remote replica
				while (!replyMessage[0].toString().toLowerCase()
						.startsWith("success")
						&& !replyMessage[0].toString().toLowerCase()
								.startsWith("sp")) {
					if (coordinatorReplicaList.keySet().size() < 2)
						break;
					if (i < 1)
						continue;
					while (!replyMessage[1].toString().toLowerCase()
							.startsWith("success")
							&& !replyMessage[1].toString().toLowerCase()
									.startsWith("sp")) {
						if (coordinatorReplicaList.keySet().size() < 2)
							break;
					}
				}
			}

			request.setRemotelyProcessd(true);
		}

	}

	private class UDPClient implements Runnable {
		private Request request;
		private String hostName;
		private String UDPPort;
		private StringBuilder replyMessage;
		private String key;

		public UDPClient(Request request, String hostName, String UDPPort,
				StringBuilder replyMessage, String key) {
			super();
			this.request = request;
			this.hostName = hostName;
			this.UDPPort = UDPPort;
			this.replyMessage = replyMessage;
			this.key = key;
		}

		@Override
		public void run() {

			DatagramSocket socket = null;

			try {
				socket = new DatagramSocket();
				InetAddress address = InetAddress.getByName(hostName);
				String requestString = request.createPacketString();

				DatagramPacket requestPacket = new DatagramPacket(
						requestString.getBytes(), requestString.length(),
						address, new Integer(UDPPort));

				// sending request to the other replica. If no response it will
				// try second time and then will consider request completed
				for (int i = 0; i < 2; i++) {
					socket.send(requestPacket);
					System.out.println("Request sent from "
							+ myCoordinator.myReplicaName + " to " + key + ": "
							+ request.getRequestID());
					socket.setSoTimeout(2000);

					byte[] buffer = new byte[1000];
					DatagramPacket reply = new DatagramPacket(buffer,
							buffer.length);

					try {
						socket.receive(reply);
						replyMessage = replyMessage.replace(0,
								replyMessage.length(),
								new String(reply.getData()).trim());

						System.out.println(" Reply received from " + key + ": "
								+ replyMessage);
						break;
					} catch (SocketTimeoutException e) {
						replyMessage = replyMessage.replace(0,
								replyMessage.length(), "Success");
						System.out.println(" No reply message from " + key);
					}
				}

			} catch (SocketException e) {
				e.printStackTrace();
			} catch (UnknownHostException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				if (socket != null)
					socket.close();
			}
		} // while ends;
	}

}
